package com.sdkj.seckill.rocketmqTest.orderMessage;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Description 顺序消费消息，带事务方式（应用可控制offset什么时候提交）
 * @Author Jiang
 * @Date 2022/01/18 19:50
 * @Version 1.0.0
 */
/**
 * 如下展示的实例是分区有序，一个订单的顺序流程是：创建、付款、推送、完成。
 * 订单号相同的消息会被先后发送到同一个队列中，消费时，同一个OrderId获取到的肯定是同一个队列。
 */
public class ConsumerInOrder {

    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
        consumer.setNamesrvAddr("47.94.172.125:9876");
        /**
         * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
         * 如果非第一次启动，那么按照上次消费的位置继续消费
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe("TopicTest", "TagA || TagC || TagD");

        consumer.registerMessageListener(new MessageListenerOrderly() {

            Random random = new Random();

            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgSet, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                for (MessageExt msg : msgSet) {
                    //可以看到每个queue有唯一的consume线程来消费，订单对每个queue（分区）有序
                    System.out.println("consumeThread=" + Thread.currentThread().getName() +
                            "queueId=" + msg.getQueueId() + ",content:" + new String(msg.getBody()));

                }
                try {
                    //模拟业务处理中
                    TimeUnit.SECONDS.sleep(random.nextInt(10));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();

        System.out.printf("Consumer Started.%n");

    }
}
